package com.zhiyuancode.mqack.worker.checker;


import com.zhiyuancode.mqack.message.base.MqResendErrorCode;
import com.zhiyuancode.mqack.message.base.MqResendRuntimeException;
import com.zhiyuancode.mqack.message.base.addressAndMessage.AddressAndMessage4Object;
import com.zhiyuancode.mqack.message.base.addressAndMessage.BaseAddressAndMessage;
import com.zhiyuancode.mqack.worker.RedisOperationsContext;
import com.zhiyuancode.mqack.worker.Sender;
import com.zhiyuancode.mqack.worker.recorder.Recorder4Object;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author 杨志远
 * email faryangsh@163.com
 * Date    2018-05-02
 */
@Slf4j
public class Checker4Object extends BaseChecker {
    private RedisOperationsContext redisOperationsContext;

    public Checker4Object(Recorder4Object recorder4Object, RedisOperationsContext redisOperationsContext) {
        this.recorder = recorder4Object;
        this.redisOperationsContext = redisOperationsContext;
    }

    public List<AddressAndMessage4Object> checkTaskTimer() throws IOException {
        if (!redisOperationsContext.checkServerHealth()) {
            return null;
        }

        log.info("start of checking message received");
        List<? extends BaseAddressAndMessage> mqAddressAndMessages = getMqAddressAndMessages();
        if (mqAddressAndMessages == null || mqAddressAndMessages.size() == 0) {
            log.info("end of checking message received");
            return null;
        }

        List<AddressAndMessage4Object> mqList = mqAddressAndMessages.stream()
                .map(e -> (AddressAndMessage4Object) e)
                .collect(Collectors.toList());
        if (mqList != null) {
            log.info("resend message: total number={},class name={}", mqList.size(), recorder.getMessageClass().getName());
            doResend(mqList);
        }

        log.info("end of checking message received");
        return mqList;
    }

    protected void doResend(List<AddressAndMessage4Object> mqList) {
        mqList.forEach(mqAddressAndMessage4Object -> {
            log.info("resend message: address={},content={}", mqAddressAndMessage4Object.getMqAddress(), mqAddressAndMessage4Object.getMessage().toJson());
            try {
                Sender.sendMessage(mqAddressAndMessage4Object);
            } catch (Exception e) {
                throw new MqResendRuntimeException(MqResendErrorCode.MQ_CONNECTION_EXCEPTION, e);
            }

        });
    }
}
